package com.bw.dws;

import com.bw.bean.UserBehavior;
import com.bw.util.ProfileCalcUtil;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.sql.Timestamp;
import java.util.HashMap;
import java.util.Map;

/**
 * 年龄标签实时计算（多维度加权）
 * 文档参考：🔶1-30（多行为加权判定）、🔶1-32（类目系数）、🔶1-90（维度权重）
 */
public class AgeTagCalc extends KeyedProcessFunction<String, UserBehavior, Map<String, Object>> {
    // 1. 各维度系数映射（文档表数据）
    private static final Map<String, Map<String, Double>> CATEGORY_COEF = new HashMap<>(); // 类目-年龄段系数
    private static final Map<String, Map<String, Double>> BRAND_COEF = new HashMap<>();    // 品牌-年龄段系数
    private static final Map<String, Map<String, Double>> PRICE_COEF = new HashMap<>();    // 价格-年龄段系数

    // 初始化系数（文档表数据，如🔶1-32：潮流服饰18-24岁系数0.9）
    static {
        // 类目系数：key=类目名，value=年龄段→系数
        Map<String, Double> fashionCoef = new HashMap<>();
        fashionCoef.put("18-24岁", 0.9);
        fashionCoef.put("25-29岁", 0.8);
        fashionCoef.put("30-34岁", 0.6);
        fashionCoef.put("35-39岁", 0.4);
        fashionCoef.put("40-49岁", 0.2);
        fashionCoef.put("50岁以上", 0.1);
        CATEGORY_COEF.put("潮流服饰", fashionCoef);

        Map<String, Double> homeCoef = new HashMap<>();
        homeCoef.put("18-24岁", 0.2);
        homeCoef.put("25-29岁", 0.4);
        homeCoef.put("30-34岁", 0.6);
        homeCoef.put("35-39岁", 0.8);
        homeCoef.put("40-49岁", 0.9);
        homeCoef.put("50岁以上", 0.7);
        CATEGORY_COEF.put("家居用品", homeCoef);

        // 品牌系数（如🔶1-33：ZARA 18-24岁系数0.9）
        Map<String, Double> zaraCoef = new HashMap<>();
        zaraCoef.put("18-24岁", 0.9);
        zaraCoef.put("25-29岁", 0.7);
        zaraCoef.put("30-34岁", 0.5);
        zaraCoef.put("35-39岁", 0.3);
        zaraCoef.put("40-49岁", 0.2);
        zaraCoef.put("50岁以上", 0.1);
        BRAND_COEF.put("ZARA", zaraCoef);

        // 价格系数（如🔶1-39：低价18-24岁系数0.8）
        Map<String, Double> lowPriceCoef = new HashMap<>();
        lowPriceCoef.put("18-24岁", 0.8);
        lowPriceCoef.put("25-29岁", 0.6);
        lowPriceCoef.put("30-34岁", 0.4);
        lowPriceCoef.put("35-39岁", 0.3);
        lowPriceCoef.put("40-49岁", 0.2);
        lowPriceCoef.put("50岁以上", 0.1);
        PRICE_COEF.put("low", lowPriceCoef);
    }

    // 2. 状态存储：各年龄段累计得分（key=年龄段，value=累计得分）
    private ValueState<Map<String, BigDecimal>> ageScoreState;

    @Override
    public void open(Configuration parameters) throws Exception {
        // 初始化状态（TTL=30天，对应文档"近30天行为"）
        ValueStateDescriptor<Map<String, BigDecimal>> descriptor = new ValueStateDescriptor<>(
                "ageScoreState",
                org.apache.flink.api.common.typeinfo.Types.MAP(
                    org.apache.flink.api.common.typeinfo.Types.STRING,
                    org.apache.flink.api.common.typeinfo.Types.BIG_DEC
                )
        );
        descriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(30)).build());
        ageScoreState = getRuntimeContext().getState(descriptor);
    }

    @Override
    public void processElement(UserBehavior behavior, Context ctx, Collector<Map<String, Object>> out) throws Exception {
        // （1）初始化状态：若首次行为，初始化6个年龄段得分
        Map<String, BigDecimal> scoreMap = ageScoreState.value();
        if (scoreMap == null) {
            scoreMap = new HashMap<>();
            scoreMap.put("18-24岁", BigDecimal.ZERO);
            scoreMap.put("25-29岁", BigDecimal.ZERO);
            scoreMap.put("30-34岁", BigDecimal.ZERO);
            scoreMap.put("35-39岁", BigDecimal.ZERO);
            scoreMap.put("40-49岁", BigDecimal.ZERO);
            scoreMap.put("50岁以上", BigDecimal.ZERO);
        }

        // （2）计算当前行为的各维度得分（行为权重×维度系数×维度权重）
        double behaviorWeight = ProfileCalcUtil.getBehaviorWeight(behavior.getBehaviorType());
        String category = behavior.getCategoryName();
        String brand = behavior.getBrandName();
        String price = behavior.getPriceRange();

        // 维度权重（文档🔶1-90：类目30%、品牌20%、价格15%，其他维度简化为0.05）
        final double CATEGORY_WEIGHT = 0.3;
        final double BRAND_WEIGHT = 0.2;
        final double PRICE_WEIGHT = 0.15;

        // 遍历所有年龄段，累加得分
        for (String ageGroup : scoreMap.keySet()) {
            BigDecimal currentScore = scoreMap.get(ageGroup);

            // 类目得分
            double categoryCoef = CATEGORY_COEF.getOrDefault(category, new HashMap<>()).getOrDefault(ageGroup, 0.0);
            BigDecimal categoryScore = BigDecimal.valueOf(behaviorWeight * categoryCoef * CATEGORY_WEIGHT);

            // 品牌得分
            double brandCoef = BRAND_COEF.getOrDefault(brand, new HashMap<>()).getOrDefault(ageGroup, 0.0);
            BigDecimal brandScore = BigDecimal.valueOf(behaviorWeight * brandCoef * BRAND_WEIGHT);

            // 价格得分
            double priceCoef = PRICE_COEF.getOrDefault(price, new HashMap<>()).getOrDefault(ageGroup, 0.0);
            BigDecimal priceScore = BigDecimal.valueOf(behaviorWeight * priceCoef * PRICE_WEIGHT);

            // 总得分累加（其他维度如时间/搜索词可按文档扩展）
            BigDecimal totalAdd = categoryScore.add(brandScore).add(priceScore);
            scoreMap.put(ageGroup, currentScore.add(totalAdd));
        }

        // （3）更新状态
        ageScoreState.update(scoreMap);

        // （4）输出：取得分最高的年龄段（文档🔶1-107逻辑）
        String bestAgeGroup = scoreMap.entrySet().stream()
                .max(Map.Entry.comparingByValue())
                .map(Map.Entry::getKey)
                .orElse("未知");

        Map<String, Object> result = new HashMap<>();
        result.put("userId", behavior.getUserId());
        result.put("ageGroup", bestAgeGroup);
        result.put("updateTime", new Timestamp(System.currentTimeMillis()));
        out.collect(result);
    }
}